package org.shanktech.tshare.binlogsync.service.impl;

import org.shanktech.tshare.binlogsync.engine.TransferEngine;
import org.shanktech.tshare.binlogsync.listener.NotificationListener;
import org.shanktech.tshare.binlogsync.model.api.StartRequestParam;
import org.shanktech.tshare.binlogsync.service.ManagerService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import com.google.code.or.OpenReplicator;

@Service
public class ManagerServiceImpl implements ManagerService {

  @Autowired
  private OpenReplicator openReplicator;

  private TransferEngine transferEngine;

  @Override
  public void startOpenReplicator(StartRequestParam param) throws Exception {
    if (null != param) {
      resetOpenReplicatorParam(param);
    }

    // 启动mysql数据库binlog日志监听
    openReplicator.start();

    // 启动转换日志线程池
    this.transferEngine = new TransferEngine();
    this.transferEngine.start();
  }

  @Override
  public void stopOpenReplicator() throws Exception {
    openReplicator.stop(0, null);
    this.transferEngine.stop();
  }

  private void resetOpenReplicatorParam(StartRequestParam param) {

    // host
    if (!StringUtils.isEmpty(param.getHost())) {
      openReplicator.setHost(param.getHost());
    }

    // port
    if (param.getPort() > 0) {
      openReplicator.setPort(param.getPort());
    }

    // user
    if (!StringUtils.isEmpty(param.getUser())) {
      openReplicator.setUser(param.getUser());
    }

    // Password
    if (!StringUtils.isEmpty(param.getPassword())) {
      openReplicator.setPassword(param.getPassword());
    }

    // ServerId
    if (param.getServerId() > -1) {
      openReplicator.setServerId(param.getServerId());
    }

    // BinlogFileName
    if (!StringUtils.isEmpty(param.getBinlogFileName())) {
      openReplicator.setBinlogFileName(param.getBinlogFileName());
    }

    // BinlogPositio
    if (param.getBinlogPosition() > 0) {
      openReplicator.setBinlogPosition(param.getBinlogPosition());
    }

    if (null != param.getDatabases()) {
      ((NotificationListener) openReplicator.getBinlogEventListener())
          .setDataBases(param.getDatabases());
    }

    if (null != param.getTables()) {
      ((NotificationListener) openReplicator.getBinlogEventListener())
          .setTables(param.getTables());
    }
  }

  public TransferEngine getTransferEngine() {
    return transferEngine;
  }

  public void setTransferEngine(TransferEngine transferEngine) {
    this.transferEngine = transferEngine;
  }

}
